package com.gugee.tiktok.data.consumer.music.core.service;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.tiktok.common.model.dal.blogger.BloggerTypeRepository;
import com.gugee.tiktok.common.model.dal.music.MusicRecordRepository;
import com.gugee.tiktok.common.model.dal.music.MusicRepository;
import com.gugee.tiktok.common.model.dal.music.MusicTypeRelativeRepository;
import com.gugee.tiktok.common.model.es.music.ESCategory;
import com.gugee.tiktok.common.model.es.music.ESDayVideoGrowth;
import com.gugee.tiktok.common.model.es.music.ESMusic;
import com.gugee.tiktok.common.model.es.music.ESVideo;
import com.gugee.tiktok.common.model.mysql.*;
import com.gugee.tiktok.common.model.ranking.GrowthData;
import com.gugee.tiktok.common.model.spider.aweme.BloggerAwemesResponse;
import com.gugee.tiktok.common.model.spider.blogger.BloggerResponse;
import com.gugee.tiktok.common.model.spider.music.MusicResponse;
import com.gugee.tiktok.data.common.msg.MusicMessage;
import com.gugee.tiktok.data.consumer.music.monitor.AopUtil;
import com.gugee.tiktok.data.consumer.music.config.MusicAppConfig;
import com.gugee.tiktok.data.consumer.music.core.mapper.MusicMapper;
import com.gugee.tiktok.data.consumer.music.core.mapper.MusicRecordMapper;
import com.gugee.tiktok.data.consumer.music.core.spider.MusicSpiderClient;

import com.gugee.tiktok.data.consumer.music.monitor.Monitor;
import com.xiguaji.boot.filecenter.FileTemplate;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import io.micrometer.core.instrument.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;


@Service
@Slf4j
public class MusicService {

    private static String MUSIC_INDEX = "music_idx";
    @Autowired
    MusicSpiderClient musicSpiderClient;

    @Autowired
    MusicAppConfig musicAppConfig;

    /*@Autowired
    TikCodeSpiderClient tikCodeSpiderClient;*/

    @Autowired
    MusicRecordRepository musicRecordRepository;

    @Autowired
    MusicRepository musicRepository;

    @Autowired
    MusicTypeRelativeRepository musicTypeRelativeRepository;

    @Autowired
    BloggerTypeRepository bloggerTypeRepository;

    @Autowired
    MusicMapper musicMapper;

    @Autowired
    MusicRecordMapper musicRecordMapper;

    @Autowired
    RestHighLevelClient client;

    @Autowired
    ESMusic.ESMusicMapping esMusicMapping;

    @Autowired
    FileTemplate fileTemplate;

    @Value("${xiguaji.oss.bucket.tiktok.videoImg}")
    String picBucket;


    Map<String, BloggerType> typeMap;

    Map<Long, BloggerType> typeMapById;


    @Autowired
    ScheduleTemplate scheduleTemplate;

    ForkJoinPool forkJoinPool;

    // ForkJoinPool forkJoinPool;

    @PostConstruct
    private void postConstruct(){
        List<BloggerType> types = bloggerTypeRepository.findAll();
        typeMap = new HashMap<>();
        typeMapById = new HashMap<>();
        types.forEach( t-> typeMap.put(t.getName(),t));
        types.forEach( t-> typeMapById.put(t.getId(),t));
        forkJoinPool = ThreadPoolUtils.forkJoinPool(musicAppConfig.getCorePollSize(), "RefreshMusic work");
        scheduleTemplate.attach("RefreshMusic work",forkJoinPool);
        // customPoolMain = scheduleTemplate.getCustomThreadPool(updateMusicMain, corePollSize, corePollSize, 0, queueSize, TimeUnit.MINUTES);
        // forkJoinPool = ThreadPoolUtils.forkJoinPool(corePollSize, "RefreshMusic work");
    }


    
    @Monitor(value = "批量更新音乐数据",maxTime = 1,timeUnit = Monitor.TimeUnit.MINUTE)
    public void updateMusic(List<MusicMessage> musicMessages){
        log.info("音乐过滤前的长度：{}",musicMessages.size());
        long start = System.currentTimeMillis()/1000;
        List<Music> musics = new ArrayList<>();
        HashSet<Long> musicIds = new HashSet<>();

        musicMessages.forEach( musicMessage -> {
            if(musicMessage.getMid() > 0){
                musicIds.add(musicMessage.getMid());
            }
        });

        log.info("准备更新音乐数据，共{}条",musicIds.size());


        /*ForkJoinPool forkJoinPool = ThreadPoolUtils.forkJoinPool(corePollSize, "RefreshMusic work");
        try {
            forkJoinPool.submit(() ->
                    musicMessages.parallelStream().forEach(m -> {
                        try {
                            musics.add(AopUtil.getSelf(this).updateMusic(m.getMid(),m.getId()));
                        } catch (Exception ex) {
                            log.error("音乐更新出错!id:{};mid:{}",m.getId(),m.getMid(),ex);
                        }
                    })
            ).get();
        }catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }finally {
            forkJoinPool.shutdown();
        }*/

        try {
            forkJoinPool.submit(() ->
                    musicIds.parallelStream().forEach(m -> {
                        try {
                            musics.add(AopUtil.getSelf(this).updateMusic(m));
                        } catch (Exception ex) {
                            log.error("音乐更新出错!mid:{}",m,ex);
                        }
                    })
            ).get();
        } catch (InterruptedException e) {
            log.error("音乐多线程更新出错!",e);
        } catch (ExecutionException e) {
            log.error("音乐多线程更新出错!",e);
        }




        // 批量保存mysql
        List<Music> completeMusic = AopUtil.getSelf(this).saveMusic(musics);

        List<ESMusic> esMusics = esMusicMapping.mappger(completeMusic);

        // 批量保存ES
        AopUtil.getSelf(this).saveToES(esMusics);
        log.info("音乐数据更新完成，原数据为{}条，成功更新{}条,总用时{}秒",musicIds.size(),musics.size(),System.currentTimeMillis()/1000 -start);

    }

    /**
     * 更新音乐
     * @param musicId
     */

    @Monitor(value = "访问接口获取音乐详情")
    public Music updateMusic(long musicId){
        MusicResponse musicResponse = AopUtil.getSelf(this).spiderMusicDetail(musicId);
        Music music = musicMapper.mapper(musicResponse.getMusic_info());
        Optional<Music> musicOptional = musicRepository.findByMid(musicId);

        if(musicOptional.isPresent()){
            music.setRegion(musicOptional.get().getRegion());
            music.setId(musicOptional.get().getId());
            music.setQrCode(musicOptional.get().getQrCode());
            music.setCreateBy(musicOptional.get().getCreateBy());
            music.setCreateTime(musicOptional.get().getCreateTime());
        }else{
            music.setCreateBy("add music data");
            music.setCreateTime(System.currentTimeMillis()/1000);
        }
        music.setUpdateBy("update music data");
        music.setUpdateTime(System.currentTimeMillis()/1000);


        // 归属国家
        if(StringUtils.isEmpty(music.getRegion())){
            if(null != music.getUid() && music.getUid() != 0){
                BloggerResponse response = AopUtil.getSelf(this).spiderBloggerDetail(music.getUid());
                music.setRegion(response.getUser().getRegion());
            }else {
                music.setRegion(getRegion(musicId));
            }
        }

        // 二维码
        /*if(StringUtils.isEmpty(music.getQrCode())){
            music.setQrCode(AopUtil.getSelf(this).getQrCode(musicId));
        }else{
            music.setQrCode(musicOptional.get().getQrCode());
        }*/

        // 打点记录并获取增量值
        saveRecord(musicId,music.getUserCount());
        val avgGrowth = getAvgGrowth(musicId);
        music.setUserGrowth(avgGrowth.getUserGrowth());

        // 七天增量值
        music.setDayVideoGrowth(JsonUtil.parseJson(getVideoDayGrowth(musicId)));

        // top3视频
        music.setTopVideos(JsonUtil.parseJson(getTopAweme(musicId)));

        return music;
    }


    public List<Music> saveMusic(List<Music> musics){
        List<Music> newMusic = new ArrayList<>();
        List<Music> updateMusic = new ArrayList<>();

        musics.forEach( music -> {
            if(null == music.getId()) {
                newMusic.add(music);
            }else {
                updateMusic.add(music);
            }
        });

        if(updateMusic.size() > 0){
            AopUtil.getSelf(this).batchUpdateMusic(updateMusic);

        }

        if(newMusic.size() > 0){
            AopUtil.getSelf(this).batchInsertMusic(newMusic);
            return musicRepository.findByMid(musics);
        }
        return updateMusic;

    }

    @Monitor(value = "批量插入音乐详情到MYSQL",maxTime = 5)
    public void batchInsertMusic(List<Music> musics){
        // 批量保存mysql
        long start = System.currentTimeMillis()/1000;

        try{
            musicRepository.batchInsert(musics);
        }catch (Exception e){
            log.error("音乐批量保存异常，使用单独保存");

            musics.parallelStream().forEach(music -> {
                try{
                    musicRepository.save(music);
                }catch (Exception ex){
                    log.error("error save music mid:{}",music.getMid());
                }
            });

        }

        if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) log.info("手动打印批量插入音乐详情到MYSQL,数量：{},用时：{}秒",musics.size(),System.currentTimeMillis()/1000 -start);
    }


    @Monitor(value = "批量更新音乐详情到MYSQL",maxTime = 5)
    public void batchUpdateMusic(List<Music> musics){
        // 批量保存mysql
        long start = System.currentTimeMillis()/1000;
        musicRepository.batchUpdate(musics);
        if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) log.info("手动打印批量更新音乐详情到MYSQL,数量：{},用时：{}秒",musics.size(),System.currentTimeMillis()/1000 -start);
    }


    @Monitor(value = "爬虫获取音乐详情",maxTime = 5)
    public MusicResponse spiderMusicDetail(long musicId){
        long start = System.currentTimeMillis()/1000;
        val musicResponse = musicSpiderClient.get(musicId);

        if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
            log.info("手动爬虫获取音乐详情,用时：{}秒",System.currentTimeMillis()/1000 -start);
        }
        return musicResponse;
    }

    @Monitor(value = "爬虫获取音乐视频列表",maxTime = 5)
    public BloggerAwemesResponse spiderVideos(long musicId){
        long start = System.currentTimeMillis()/1000;
        val awemes = musicSpiderClient.getAwemes(musicId, 3);

        if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
            log.info("手动爬虫获取音乐视频列表,用时：{}秒",System.currentTimeMillis()/1000 -start);
        }
        return awemes;
    }

    /**
     * 获取播主
     * @param musicId
     * @return
     */

    @Monitor(value = "爬虫获取播主详情",maxTime = 5)
    public BloggerResponse spiderBloggerDetail(long musicId){

        long start = System.currentTimeMillis()/1000;
        val blogger = musicSpiderClient.getBlogger(musicId, null);
        if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
            log.info("手动爬虫获取播主详情,用时：{}秒",System.currentTimeMillis()/1000 -start);
        }
        return blogger;
    }




    /**
     * 打点记录音乐增量数据并返回
     * @param musicId
     * @param userCount
     * @return
     */

    public long saveRecord(long musicId,long userCount){
        int todayCode = Integer.parseInt(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        int yesterdayCode = Integer.parseInt(LocalDateTime.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd")));

        Optional<MusicRecord> todayRecordOptional = musicRecordRepository.findByMidAndDateCode(musicId,todayCode);
        Optional<MusicRecord> yesterdayRecordOptional = musicRecordRepository.findByMidAndDateCode(musicId,yesterdayCode);
        MusicRecord musicRecord = new MusicRecord();

        if(todayRecordOptional.isPresent()){
            musicRecord = todayRecordOptional.get();
            musicRecord.setUpdateTime(System.currentTimeMillis()/1000);
            musicRecord.setUpdateBy("update record music data");
        }else {
            musicRecord.setMid(musicId);
            musicRecord.setDateCode(todayCode);
            musicRecord.setDateCode(todayCode);
            musicRecord.setCreateBy("add record music data");
            musicRecord.setUpdateBy("add record music data");
            musicRecord.setCreateTime(System.currentTimeMillis()/1000);
            musicRecord.setUpdateTime(System.currentTimeMillis()/1000);
            musicRecord.setUserGrowth(0l);
        }
        musicRecord.setUserCount(userCount);

        if(yesterdayRecordOptional.isPresent()){
            musicRecord.setUserGrowth(musicRecord.getUserCount() - yesterdayRecordOptional.get().getUserCount());
        }
        musicRecordRepository.save(musicRecord);
        return musicRecord.getUserGrowth();
    }


    /**
     * 获取二维码
     * @param musicId
     * @return
     */

    /*@Monitor(value = "获取二维码",maxTime = 5)
    public String getQrCode(long musicId){
        try{
            long start = System.currentTimeMillis()/1000;
            val qrCodeResponse = tikCodeSpiderClient.get(musicId, 3);
            if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
                log.info("手动获取二维码,用时：{}秒",System.currentTimeMillis()/1000 -start);
            }
            if(CollectionUtils.isEmpty(qrCodeResponse.getQrcode_url().getUrl_list())) {
                return "";
            }
            return qrCodeResponse.getQrcode_url().getUrl_list().get(0);
        }catch (Exception e){
            log.error("获取二维码失败，musicId:{}",musicId,e);
            return "";
        }
    }*/


    public GrowthData getAvgGrowth(long musicId){
        int todayCode = Integer.parseInt(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        int beforeCode = Integer.parseInt(LocalDateTime.now().minusDays(musicAppConfig.getAvgDay()).format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        return musicRecordRepository.findByGrowthData(musicId,beforeCode,todayCode);
    }

    /**
     * 获取国家
     * @param musicId
     * @return
     */

    public String getRegion(long musicId){
        Map<String,Object> regionMap =  musicRepository.getBelongRegion(musicId,musicAppConfig.getLatestDays());
        return regionMap.get("region") == null ?"":regionMap.get("region").toString();
    }

    /**
     * 获取七天视频增长趋势图
     * @param musicId
     * @return
     */

    public List<ESDayVideoGrowth> getVideoDayGrowth(long musicId){
        int startCode = Integer.parseInt(LocalDateTime.now().minusDays(7).format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        int afterCode = Integer.parseInt(LocalDateTime.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        // int afterCode = Integer.parseInt(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        List<MusicRecord> musicRecords = musicRecordRepository.findByMidAndDateCodeBetweenOrderByDateCode(musicId,startCode,afterCode);
        return musicRecordMapper.mapperToES(musicRecords);
    }

    /**
     * 保存到ES
     * @param esMusic
     */

    public void saveToES(ESMusic esMusic){

        IndexRequest indexRequest = new IndexRequest(MUSIC_INDEX);
        indexRequest.id(String.valueOf(esMusic.getId()));
        indexRequest.source(JsonUtil.parseJson(esMusic), XContentType.JSON);

        UpdateRequest updateRequest = new UpdateRequest(MUSIC_INDEX,String.valueOf(esMusic.getId())).
                doc(JsonUtil.parseJson(esMusic), XContentType.JSON).
                upsert(indexRequest);
        updateRequest.docAsUpsert(true);
        try{
            client.update(updateRequest, RequestOptions.DEFAULT);
        }catch (Exception e){
            log.error("save to es error",e);
        }
    }

    /**
     * 保存到ES
     * @param esMusics
     */

    @Monitor(value = "批量将音乐保存到ES",maxTime = 5)
    public void saveToES(List<ESMusic> esMusics){
        BulkRequest bulkRequest = new BulkRequest();

        try {
            forkJoinPool.submit(() ->
                    esMusics.parallelStream().forEach(m -> {
                        try {
                            //分类
                            m.setCategories(AopUtil.getSelf(this).setBelongType(m.getMid()));
                            IndexRequest indexRequest = new IndexRequest(MUSIC_INDEX);
                            indexRequest.id(String.valueOf(m.getId()));
                            indexRequest.source(JsonUtil.parseJson(m), XContentType.JSON);
                            bulkRequest.add(indexRequest);
                        } catch (Exception ex) {
                            log.error("音乐分类出错!mid:{}" , m.getMid(), ex);
                        }
                    })
            ).get();
        }catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        /*ForkJoinPool forkJoinPool2 = ThreadPoolUtils.forkJoinPool(300, "RefreshAwemeCommentEvent work");

        try {
            forkJoinPool2.submit(() ->
                    esMusics.parallelStream().forEach(m -> {
                        try {
                            //分类
                            m.setCategories(AopUtil.getSelf(this).setBelongType(m.getMid()));
                            IndexRequest indexRequest = new IndexRequest(MUSIC_INDEX);
                            indexRequest.id(String.valueOf(m.getId()));
                            indexRequest.source(JsonUtil.parseJson(m), XContentType.JSON);
                            bulkRequest.add(indexRequest);
                        } catch (Exception ex) {
                            log.error("音乐分类出错!mid:{}" , m.getMid(), ex);
                        }
                    })
            ).get();
        }catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }finally{
            forkJoinPool2.shutdown();
        }*/



        try{

            long start = System.currentTimeMillis()/1000;
            client.bulk(bulkRequest,RequestOptions.DEFAULT);
            if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
                log.info("手动批量将音乐保存到ES,数量：{},用时：{}秒",esMusics.size(),System.currentTimeMillis()/1000 -start);
            }
        }catch (Exception e){
            log.error("save to es error",e);
        }
    }

    /**
     * 获取音乐所属分类
     * @param musicId
     * @return
     */
    @Monitor(value = "音乐分类",maxTime = 5)
    public List<ESCategory> setBelongType(long musicId){
        long start = System.currentTimeMillis()/1000;
        long uesBelongType;
        long uesGetBelongType;


        List<ESCategory> esCategories = new ArrayList<>();
        List<MusicTypeRelative> musicTypeRelatives = musicTypeRelativeRepository.findByMid(musicId);

        if(musicTypeRelatives.size() > 0){
            musicTypeRelatives.forEach( musicTypeRelative -> {
                val type = typeMapById.get(musicTypeRelative.getTypeId());
                ESCategory esCategory = new ESCategory();
                esCategory.setId(type.getId());
                esCategory.setTypeName(type.getName());
                esCategories.add(esCategory);
            });
            return esCategories;
        }

        /*if(true){
            return esCategories;
        }*/

        long startBelongType = System.currentTimeMillis()/1000;
        List<Map<String,Object>> mapList = musicRepository.getBelongType(musicId,musicAppConfig.getLatestDays());
        uesBelongType = System.currentTimeMillis()/1000 - startBelongType;
        long sum = mapList.stream().mapToLong(m -> (long) m.get("count")).sum();

        long startGetBelongType = System.currentTimeMillis()/1000;
        for (int i = 0; i < mapList.size(); i++) {
            BloggerType type = typeMap.get(mapList.get(i).get("type"));
            if(null == type) continue;

            if(i > 1 ){
                BigDecimal bigDecimal = new BigDecimal((long)mapList.get(i).get("count"));
                double percent =bigDecimal.divide(new BigDecimal(sum),3, RoundingMode.FLOOR).multiply(new BigDecimal(100)).setScale(2).doubleValue();

                if(percent < 20){
                    break;
                }
            }
            MusicTypeRelative musicTypeRelative = new MusicTypeRelative();
            musicTypeRelative.setMid(musicId);
            musicTypeRelative.setCreateTime(System.currentTimeMillis()/1000);
            musicTypeRelative.setUpdateTime(System.currentTimeMillis()/1000);
            musicTypeRelative.setCreateBy("add music type");
            musicTypeRelative.setUpdateBy("add music type");
            musicTypeRelative.setTypeId(type.getId());
            // musicTypeRelatives.add(musicTypeRelative);

            ESCategory esCategory = new ESCategory();
            esCategory.setId(type.getId());
            esCategory.setTypeName(type.getName());
            esCategories.add(esCategory);
        }
        uesGetBelongType = System.currentTimeMillis()/1000 - startGetBelongType;


        long startSaveAll = System.currentTimeMillis()/1000;
        // musicTypeRelativeRepository.saveAll(musicTypeRelatives);
        if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
            log.info("手动音乐：{};分类,用时：{}秒;分类归属数据访问,用时：{}秒;分类算法归属,用时：{}秒;分类,条数：{}",
                    musicId,System.currentTimeMillis()/1000 -start,uesBelongType,uesGetBelongType,esCategories.size(),System.currentTimeMillis()/1000 -startSaveAll);
        }
        return esCategories;
    }


    public List<ESVideo> getTopAweme(long musicId){
        BloggerAwemesResponse response = AopUtil.getSelf(this).spiderVideos(musicId);
        List<ESVideo> esVideos = new ArrayList<>();

        try {
            forkJoinPool.submit(()->{
                for (int i = 0; i < response.getAweme_list().size(); i++) {
                    if(i >2){
                        break;
                    }
                    ESVideo esVideo = new ESVideo();
                    esVideo.setDiggCount(response.getAweme_list().get(i).getStatistics().getDigg_count());
                    esVideo.setCover(setAwemeCover(response.getAweme_list().get(i).getAweme_id(),response.getAweme_list().get(i).getVideo().getCover().getUrl_list().get(0)));
                    esVideo.setShareUrl(response.getAweme_list().get(i).getShare_url());
                    esVideos.add(esVideo);
                }
            }).get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return esVideos;
    }

    private String setAwemeCover(String awemeId,String coverUrl){
        String newCoverUrl = coverUrl;
        long start = System.currentTimeMillis()/1000;
        try {
            newCoverUrl = fileTemplate.putNetImg(picBucket,"video_" + DateTimeUtils.millsecond2DateStr(start,"yyyyMMdd") + "/" + awemeId,coverUrl);
        } catch (Exception e) {
            return newCoverUrl;
        }finally {
            if(System.currentTimeMillis()/1000 -start > musicAppConfig.getTimeLimit()) {
                log.info("oss上传图片用时：{}秒",
                        System.currentTimeMillis()/1000 -start);
            }
        }

        return newCoverUrl;
    }
}
